package com.lx.sync.core.executor;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.IdUtil;
import com.lx.sync.common.util.CacheUtil;
import com.lx.sync.config.model.SyncConfig;
import com.lx.sync.common.constant.Constants;
import com.lx.sync.core.constant.SyncConstant;
import com.lx.sync.core.constant.ExtDataEsInxConstant;
import com.lx.sync.core.exception.SyncFailException;
import com.lx.sync.core.model.*;
import com.lx.sync.core.service.SyncStateService;
import com.lx.sync.config.service.SyncConfigService;
import lombok.extern.slf4j.Slf4j;
import org.joda.time.DateTime;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;

import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 抽象同步服务类
 */
@Slf4j
public abstract class AbstractSyncService<T extends SyncBaseDTO> {

    protected final CacheUtil cacheUtil;
    private static final String SYNC_STATE_KEY_PREFIX = "sync:state:";
    private static final Date DEFAULT_START_DATE = DateTime.parse("1970-01-01").toDate();

    @Autowired
    protected SyncStateService syncStateService;

    @Autowired
    @Qualifier("syncConfigService")
    protected SyncConfigService syncConfigService;

//
//    @Autowired
//    protected SyncMonitor syncMonitor;

    public AbstractSyncService(CacheUtil cacheUtil) {
        this.cacheUtil = cacheUtil;
    }

    /**
     * 获取同步标志，由子类实现
     *
     * @return
     */
    protected abstract String getIdxName();

    /**
     * 由各实现类提供数据
     *
     * @param request
     * @return
     */
    protected abstract SyncPage fetchSourceData(SyncFetchDataRequest request);

    /**
     * 外部数据保存到数据库
     *
     * @param syncDataList
     */
    protected abstract void bulkToDb(List<T> syncDataList);

    /**
     * 同步方法
     */
    private long sync() {
        int count = 0;


        SyncState syncState = initSyncState();
        try {
            //1.获取同步配置
            SyncConfig syncConfig = getSyncConfig();

            //2.设置同步类型
            syncState.setSyncType(syncConfig.getSyncType());

            //3.获取同步开始时间
            Date syncStartTime = getPullStartTime(syncConfig);

            //4.获取分页模式
            boolean useCursorPagination = syncConfig.getPageMode() == 1;

            Integer executorType = syncConfig.getExecutorType();

            //5.拉取第一批数据
            SyncPage firstSyncPage;
            if (useCursorPagination) {
                firstSyncPage = fetchInitialDataByCursor(syncStartTime, syncConfig);
            } else {
                firstSyncPage = fetchInitialData(syncStartTime, syncConfig);
            }

            //6.空数据处理
            if (isNoDataToSync(firstSyncPage)) {
                handleEmptySync(syncState, syncStartTime);
                return 0L;
            }
            logSyncStart(firstSyncPage);

            List<T> docs = firstSyncPage.getRecords();


            // 处理第一批数据
            count = processFirstBatch(firstSyncPage.getRecordCount(), executorType,syncState, docs);

            // 处理剩余批次
            if (useCursorPagination) {
                count = count + processRemainingBatchesByCursor(firstSyncPage,syncConfig.getBatchSize(),executorType,syncStartTime, syncState);
            } else {
                count = count + processRemainingBatches(firstSyncPage.getTotalPages(), firstSyncPage.getPageNumber(), syncConfig.getBatchSize(), executorType, syncStartTime, syncState);
            }

            updateSyncState(syncState, count, Constants.SUCCESS, null);
            return count;
        } catch (Exception e) {
            if (!(e instanceof SyncFailException)) {
                log.error("同步失败", e);
                syncState.setFailSyncTime(getLastUpdateTime());
                updateSyncState(syncState, count, Constants.FAIL, e.getMessage());
                throw e;
            }
        }
        return count;
    }

    private SyncState initSyncState() {
        SyncState syncState = new SyncState();
        syncState.setId(IdUtil.fastSimpleUUID());
        syncState.setStartTime(new Date());
        syncState.setIdxName(getIdxName());
        return syncState;
    }

    private SyncPage fetchInitialData(Date startSyncTime, SyncConfig syncConfig) {
        SyncFetchDataRequest request = new SyncFetchDataRequest(
                startSyncTime,
                SyncConstant.START_PAGE_NUM,
                syncConfig.getBatchSize(), true
        );
        return fetchSourceData(request);
    }

    private SyncPage fetchInitialDataByCursor(Date startSyncTime, SyncConfig syncConfig) {
        SyncFetchDataRequest request = new SyncFetchDataRequest(
                startSyncTime,
                null,
                syncConfig.getBatchSize()
        );
        return fetchSourceData(request);
    }

    private boolean isNoDataToSync(SyncPage syncPage) {
        return Objects.isNull(syncPage) || CollectionUtil.isEmpty(syncPage.getRecords());
    }

    private void handleEmptySync(SyncState syncState, Date startSyncTime) {
        syncState.setSyncCount(0);
        syncState.setSumCount(0);
        syncState.setState(Constants.SUCCESS);
        syncState.setErrorMessage(null);
        syncState.setEndTime(new Date());
        syncState.setEarliestSyncTime(startSyncTime);
        syncState.setLatestSyncTime(startSyncTime);
        saveSyncState(syncState);
    }

    private void logSyncStart(SyncPage syncPage) {
        log.info(">>> 同步 [{}], 总条数 :: {}", getIdxName(), syncPage.getRecordCount());
    }

    private int processFirstBatch(  int totalCount,Integer executorType,SyncState syncState,List<T> docs) {
        try {
            executeSync(executorType,docs);
            syncState.setSumCount(totalCount);
            syncState.setSyncCount(docs.size());
            syncState.setState(Constants.IN_PROGRES);
            syncState.setErrorMessage(null);
            syncState.setEarliestSyncTime(docs.get(0).getUpdateTime());
            syncState.setLatestSyncTime(docs.get(docs.size() - 1).getUpdateTime());
            saveSyncState(syncState);
            return docs.size();
        } catch (Exception e) {
            handleSyncError(syncState, docs, e);
            throw new SyncFailException(e.getMessage());
        }
    }

    private void handleSyncError(SyncState syncState, List<T> docs, Exception e) {
        syncState.setSyncCount(0);
        syncState.setSumCount(0);
        syncState.setState(Constants.FAIL);
        syncState.setErrorMessage(e.getMessage());
        syncState.setEndTime(new Date());
        syncState.setFailSyncTime(docs.get(0).getUpdateTime());
        saveSyncState(syncState);
    }

    private int processRemainingBatches(int totalPages, int pageNumber, int batchSize,Integer executorType, Date startSyncTime,SyncState syncState) {
        int count = 0;
        SyncPage syncPage = initRemainingBatchesSyncPage(totalPages, pageNumber, batchSize);
        while (syncPage.getPageNumber() < totalPages) {
            syncPage.setPageNumber(syncPage.getPageNumber() + 1);
            logBatchProcessing(syncPage);
            syncPage = fetchNextPage(batchSize, startSyncTime, syncPage);
            if (Objects.isNull(syncPage) || CollectionUtil.isEmpty(syncPage.getRecords())) {
                log.warn(">>> 拉取外部数据为空 :: [{}]", getIdxName());
                break;
            }

            List<T> docs = syncPage.getRecords();
            try {
                syncState.setLatestSyncTime(docs.get(docs.size() - 1).getUpdateTime());
                executeSync(executorType,docs);
                count += docs.size();
            } catch (Exception e) {
                syncState.setFailSyncTime(docs.get(0).getUpdateTime());
                updateSyncState(syncState, count, Constants.FAIL, e.getMessage());
                throw new SyncFailException(e.getMessage());
            }
        }
        return count;
    }

    private int processRemainingBatchesByCursor(SyncPage currentPage,int batchSize, Integer executorType,Date startSyncTime, SyncState syncState) {
        int count = 0;
        SyncPage syncPage = currentPage;
        Long lastId = null;
        Date lastUpdateTime = startSyncTime;

        while (true) {
            if (syncPage.getRecords().isEmpty()) {
                break;
            }

            // 获取最后一条记录的ID和更新时间
            T lastRecord = (T) syncPage.getRecords().get(syncPage.getRecords().size() - 1);
            lastId = lastRecord.getId();
            lastUpdateTime = lastRecord.getUpdateTime();

            // 获取下一页数据
            SyncFetchDataRequest request = new SyncFetchDataRequest(
                    lastUpdateTime,
                    lastId,
                    batchSize
            );
            syncPage = fetchSourceData(request);

            if (Objects.isNull(syncPage) || CollectionUtil.isEmpty(syncPage.getRecords())) {
                log.warn(">>> 拉取外部数据为空 :: [{}]", getIdxName());
                break;
            }

            List<T> docs = (List<T>) syncPage.getRecords();
            try {
                syncState.setLatestSyncTime(docs.get(docs.size() - 1).getUpdateTime());
                executeSync(executorType,docs);
                count += docs.size();
            } catch (Exception e) {
                syncState.setFailSyncTime(docs.get(0).getUpdateTime());
                updateSyncState(syncState, count, Constants.FAIL, e.getMessage());
                throw new SyncFailException(e.getMessage());
            }
        }
        return count;
    }

    private SyncPage initRemainingBatchesSyncPage(int totalPages, int pageNumber, int batchSize) {
        SyncPage syncPage = new SyncPage();
        syncPage.setPageNumber(pageNumber);
        syncPage.setPageSize(batchSize);
        syncPage.setTotalPages(totalPages);
        return syncPage;
    }

    private void logBatchProcessing(SyncPage syncPage) {
        log.info(">>> 拉取外部数据 :: [{}] 开始执行, 当前页 :: [{}], 当前拉取条数 :: [{}], 满足记录总行数 :: [{}], 总页数 :: [{}]",
                getIdxName(), syncPage.getPageNumber(), syncPage.getPageSize(),
                syncPage.getRecordCount(), syncPage.getTotalPages());
    }

    private SyncPage fetchNextPage(int batchSize, Date startSyncTime, SyncPage page) {
        SyncFetchDataRequest request = new SyncFetchDataRequest(
                startSyncTime,
                page.getPageNumber(),
                batchSize, false
        );
        return fetchSourceData(request);
    }

    public final void pullExternalDataAndSync() {
        String index = getIdxName();
        log.info(">>> 开始同步数据, 类型名称 :: [{}], 缓存 key :: [{}]", index, getCacheKey());
        try {
            log.info(">>> 获取的最后更新时间 :: [{}]", getFormatLastUpdateTime());
            long startTime = System.currentTimeMillis();
            long count = sync();
            log.info(">>> 共处理 [{}] 条数据, 记录的最后更新时间 :: [{}]", count, getFormatLastUpdateTime());
            log.info(">>> 同步完成, 类型名称 :: [{}], 耗时 :: [{}] ms", index, System.currentTimeMillis() - startTime);
        } catch (Exception e) {
            log.error(String.format(">>> 同步外部数据异常, 类型名称 :: [%s]", index), e);
        }
    }



    /**
     * 组装缓存的key
     *
     * @return
     */
    private String getCacheKey() {
        return ExtDataEsInxConstant.REDIS_SYNC_PREFIX + getIdxName();
    }

    private String getSyncStateKey() {
        return SYNC_STATE_KEY_PREFIX + getIdxName();
    }

    /**
     * 获取最后更新时间
     * 如果是第一次 默认为  1970-01-01 00:00:00
     * 如果最新一次是失败的话，取最新的失败时间
     * 如果最新一次是成功的话,去最新数据同步时间
     */
    private Date getLastUpdateTime() {
        try {
            SyncState lastState = syncStateService.getLatestSyncState(getIdxName());
            if (lastState == null) {
                return DEFAULT_START_DATE;
            }

            //上一次同步状态是成功并且最后更新时间不为空，取上一次更新时间
            if (lastState.getState() == Constants.SUCCESS && lastState.getLatestSyncTime() != null) {
                return lastState.getLatestSyncTime();
            }
            //上一次同步失败并且最后同步失败的时间不为空，取上一次失败时间
            if (lastState.getState() == Constants.FAIL && lastState.getFailSyncTime() != null) {
                return lastState.getFailSyncTime();
            }

            return DEFAULT_START_DATE;
        } catch (Exception e) {
            log.error("获取同步状态失败", e);
            return null;
        }
    }

    private String getLastSyncType() {

        return "INCREMENTAL";
    }

    /**
     * 获取格式化最后更新时间
     *
     * @return
     */
    private String getFormatLastUpdateTime() {
        return DateUtil.format(getLastUpdateTime(), DatePattern.NORM_DATETIME_PATTERN);
    }

    /**
     * 保存最后更新时间
     *
     * @param docs
     */
    private void saveLastUpdateTime(List<? extends SyncBaseDTO> docs) {
        if (docs.isEmpty()) {
            return;
        }
        String index = new StringBuilder(ExtDataEsInxConstant.REDIS_SYNC_PREFIX)
                .append(getIdxName())
                .toString();

        Date newestTime = docs.get(docs.size() - 1).getUpdateTime();
        log.info(">>> 同步数据 :: [{}], 记录最后更新时间 :: [{}]",
                index, DateUtil.format(newestTime, DatePattern.NORM_DATETIME_PATTERN));

        cacheUtil.set(getCacheKey(), newestTime, 365, TimeUnit.DAYS);
    }


    private void saveSyncState(SyncState syncState) {
        // 保存同步状态
        try {
            syncStateService.saveSyncState(syncState);
        } catch (Exception e) {
            log.error("保存同步状态失败", e);
        }
    }

    /**
     * 更新同步状态
     */
    private void updateSyncState(SyncState syncState, int count, int state, String errorMessage) {
        syncState.setEndTime(new Date());
        syncState.setSyncCount(count);
        syncState.setState(state);
        syncState.setErrorMessage(errorMessage);
        syncState.setIdxName(getIdxName());

        // 保存同步状态
        try {
            syncStateService.updateSyncState(syncState);
        } catch (Exception e) {
            log.error("保存同步状态失败", e);
        }
    }

    /**
     * 存储数据至ES或DB, 累计操作数据数量, 更新最后更新时间
     *
     * @param docs
     * @return
     */
    private void executeSync(Integer executorType,List<T> docs) {
        SyncExecutor executor = ExtConfiguration.newSyncExecutor(executorType,this);
        executor.executor(docs);

    }

    /**
     * 获取同步配置
     */
    private SyncConfig getSyncConfig() {
        SyncConfig syncConfig = null;
        try {
            syncConfig = syncConfigService.getSyncConfig(getIdxName());
        } catch (Exception e) {
            log.error("同步数据失败,获取同步配置异常", e);
            throw new SyncFailException("同步数据失败,获取同步配置异常");
        }

        if (syncConfig == null) {
            throw new SyncFailException("同步数据失败,同步配置为空");
        }
        return syncConfig;

    }

    /**
     * 判断是否需要全量同步
     */
    private Date getPullStartTime(SyncConfig syncConfig) {


        // 1. 检查配置中的全量同步标志
        if (syncConfig.getSyncType().equals("FULL")) {
            return DEFAULT_START_DATE;
        }

        //2.是否使用配置中的时间
        if (syncConfig.isUseConfigPullTime()) {
            return syncConfig.getSyncStartTime();
        }

        //3. 读取上次同步时间
        return getLastUpdateTime();
    }
}
